package com.sunday.common.reactor.e10_Synchronous_World;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * A.8. 回到同步世界
 */
@Slf4j
public class SynchronousWorld {

    /**
     * A.8. 回到同步世界
     * 注意：如果从 在标记为“非阻塞”的调度程序中（默认为 parallel（） 和 single（））。
     * 我有一个 Flux<T>我想：
     * 块，直到我能得到第一个元素：Flux#blockFirst
     * ...​超时：Flux#blockFirst（Duration）
     * 块，直到我能得到最后一个元素（如果为空，则为 null）：Flux#blockLast
     * ...​超时：Flux#blockLast（Duration）
     * 同步切换到 Iterable<T>： Flux#toIterable
     * 同步切换到 Java 8 Stream<T>：Flux#toStream
     * 我有一个 Mono<T>，我想要：
     * 阻止直到我能得到值：Mono#block
     * ...​超时：Mono#block（Duration）
     * a CompletableFuture<T>： Mono#toFuture
     */

    /**
     * 直到我能得到第一个元素：Flux#blockFirst
     */
    @Test
    public void blockFirst() {
        log.info("直到我能得到第一个元素：Flux#blockFirst");
        // 直到我能得到第一个元素：Flux#blockFirst
        Flux<Integer> flux = Flux.range(1, 100);
        log.info("flux.blockFirst(): {}", flux.blockFirst());
    }

    /**
     * 超时：Flux#blockFirst（Duration）
     */
    @Test
    public void blockFirstDuration() throws InterruptedException {
        log.info("超时：Flux#blockFirst（Duration）");
        // 超时：Flux#blockFirst（Duration）
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofSeconds(2));
        // java.lang.IllegalStateException: Timeout on blocking read for 1000000000 NANOSECONDS
        log.info("flux.blockFirst(): {}", flux.blockFirst(Duration.ofSeconds(1)));
        TimeUnit.SECONDS.sleep(3);
    }

    /**
     * 直到我能得到最后一个元素（如果为空，则为 null）：Flux#blockLast
     */
    @Test
    public void blockLast() {
        log.info("直到我能得到最后一个元素（如果为空，则为 null）：Flux#blockLast");
        // 直到我能得到最后一个元素（如果为空，则为 null）：Flux#blockLast
        Flux<Integer> flux = Flux.range(1, 100);
        log.info("flux.blockLast(): {}", flux.blockLast());
    }

    /**
     * 超时：Flux#blockLast（Duration）
     */
    @Test
    public void blockLastDuration() throws InterruptedException {
        log.info("超时：Flux#blockLast（Duration）");
        // 超时：Flux#blockLast（Duration）
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofSeconds(2));
        // java.lang.IllegalStateException: Timeout on blocking read for 3000000000 NANOSECONDS
        log.info("flux.blockLast(): {}", flux.blockLast(Duration.ofSeconds(3)));
        TimeUnit.SECONDS.sleep(5);
    }

    /**
     * 同步切换到 Iterable<T>： Flux#toIterable
     */
    @Test
    public void toIterable() {
        log.info("同步切换到 Iterable<T>： Flux#toIterable");
        // 同步切换到 Iterable<T>： Flux#toIterable
        Flux<Integer> flux = Flux.range(1, 100);
        Iterable<Integer> iterable = flux.toIterable();
        iterable.forEach(i -> log.info("i: {}", i));
    }

    /**
     * 同步切换到 Java 8 Stream<T>：Flux#toStream
     */
    @Test
    public void toStream() {
        log.info("同步切换到 Java 8 Stream<T>：Flux#toStream");
        // 同步切换到 Java 8 Stream<T>：Flux#toStream
        Flux<Integer> flux = Flux.range(1, 100);
        flux.toStream().forEach(i -> log.info("i: {}", i));
    }

    /**
     * 阻止直到我能得到值：Mono#block
     */
    @Test
    public void block() {
        log.info("阻止直到我能得到值：Mono#block");
        // 阻止直到我能得到值：Mono#block
        Mono<Integer> mono = Mono.just(100);
        log.info(" mono.block(): {}", mono.block());
    }

    /**
     * 超时：Mono#block（Duration）
     */
    @Test
    public void blockDuration() throws InterruptedException {
        log.info("超时：Mono#block（Duration）");
        // 超时：Mono#block（Duration）
        Mono<Integer> mono = Mono.just(100).delayElement(Duration.ofSeconds(2));
        // java.lang.IllegalStateException: Timeout on blocking read for 1000000000 NANOSECONDS
        log.info(" mono.block(): {}", mono.block(Duration.ofSeconds(1)));
        TimeUnit.SECONDS.sleep(3);
    }

    /**
     * a CompletableFuture<T>： Mono#toFuture
     */
    @Test
    public void toFuture() throws ExecutionException, InterruptedException {
        log.info("a CompletableFuture<T>： Mono#toFuture");
        // a CompletableFuture<T>： 单声道#toFuture
        Mono<Integer> mono = Mono.just(100);
        // mono.toFuture() 返回 CompletableFuture<Integer>
        // CompletableFuture 是 JDK 1.8 引入的一个类，它实现了 Future 接口，提供了非阻塞式的异步计算能力。
        CompletableFuture<Integer> future = mono.toFuture();
        log.info("mono.toFuture(): {}", future);
        log.info("future.get(): {}", future.get());
    }

}
